package Hudi_SelfStudy.spark_shell_study

//  这些包是根据文档里面来的，如果是先写代码然后再补充的话补充不全，下面代码会报错
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.QuickstartUtils._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.spark.sql.SparkSession

import scala.collection.JavaConversions._


object study_01 {
  def main(args: Array[String]): Unit = {

    /*
          因为缺少依赖，运行会报错！！！！
          fileName:数据构造器生成数据写入虚拟机
     */



    //  准备sparkSql和hudi集成的环境
    //  serializer:序列化器,用来序列化(将对象转化为字节流),和反序列化(将字节流转化为对象)
    //  .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension"):用来启动hudi的spark扩展
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据构造器生成数据写入虚拟机的文件系统")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    //  导入隐式转换


    //  表名
    val tableName="hudi_trip_cow"
    //  将数据保存到虚拟机的hdfs上
    val savePath="hdfs://192.168.40.110:9000/user/hudi/"
    //  DataGenerator:数据生成器，用于生成数据
    val dataGen = new DataGenerator
    //  生成数据
    val inserts = convertToStringList(dataGen.generateInserts(10))
    //  将数据的格式转化为DataFrame
    val df = spark.read.json(spark.sparkContext.parallelize(inserts, 2))
    //  写入数据
    df.write.format("hudi").
      options(getQuickstartWriteConfigs).
      option(PRECOMBINE_FIELD.key(), "ts").
      option(RECORDKEY_FIELD.key(), "uuid").
      option(PARTITIONPATH_FIELD.key(), "partitionpath").
      option(TBL_NAME.key(), tableName).
      mode("overwrite").
      save(savePath)


    spark.close()

  }
}
